package com.itheima.search.listener;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.BulkRequest;
import co.elastic.clients.elasticsearch.core.BulkResponse;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import co.elastic.clients.elasticsearch.core.bulk.IndexOperation;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.itheima.search.domain.NewsDocument;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.time.OffsetDateTime;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeParseException;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * MQ全量同步消息监听器，专门处理全量数据同步到Elasticsearch
 * 使用新的Elasticsearch Java Client
 * @author NicelYHHello
 */
@Component
@Slf4j
public class NewsFullSyncListener {

    private final ObjectMapper objectMapper;
    private final ElasticsearchClient elasticsearchClient;

    // 定义日期时间格式化器
    private static final DateTimeFormatter ISO_OFFSET_DATE_TIME_FORMATTER =
            DateTimeFormatter.ISO_OFFSET_DATE_TIME;
    private static final DateTimeFormatter LOCAL_DATE_TIME_FORMATTER =
            DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    public NewsFullSyncListener(ObjectMapper objectMapper,
                                ElasticsearchClient elasticsearchClient) {
        this.objectMapper = objectMapper;
        this.elasticsearchClient = elasticsearchClient;
    }

    /**
     * 监听全量同步消息
     * 处理从数据库发送到MQ的新闻数据，同步到Elasticsearch
     */
    @RabbitListener(queues = "news.all.queue")
    public void handleFullSyncMessage(String message) {
        log.info("接收到全量同步消息，开始处理...");

        try {
            // 将JSON字符串转换为List<Map>
            List<Map<String, Object>> newsMapList = objectMapper.readValue(
                    message,
                    new TypeReference<List<Map<String, Object>>>() {}
            );

            log.info("解析到 {} 条新闻数据", newsMapList.size());

            // 将Map转换为NewsDocument对象
            List<NewsDocument> newsDocuments = newsMapList.stream()
                    .map(this::convertMapToNewsDocument)
                    .collect(Collectors.toList());

            // 批量同步到Elasticsearch
            syncToElasticsearch(newsDocuments);

            log.info("成功同步 {} 条数据到Elasticsearch", newsDocuments.size());

        } catch (Exception e) {
            log.error("处理全量同步消息失败: {}", e.getMessage(), e);
            // 可以根据需要添加重试机制或错误处理
        }
    }

    /**
     * 将Map转换为NewsDocument对象
     * 确保字段类型兼容，避免反序列化问题
     */
    private NewsDocument convertMapToNewsDocument(Map<String, Object> map) {
        NewsDocument document = new NewsDocument();

        // 手动映射每个字段，确保类型安全
        if (map.get("news_id") != null) {
            document.setNewsId(Long.valueOf(map.get("news_id").toString()));
        }
        document.setTitle((String) map.get("title"));
        if (map.get("type_id") != null) {
            document.setTypeId(Integer.valueOf(map.get("type_id").toString()));
        }
        document.setContent((String) map.get("content"));
        if (map.get("delay_time") != null) {
            document.setDelayTime(Byte.valueOf(map.get("delay_time").toString()));
        }
        document.setSource((String) map.get("source"));
        document.setCreateBy((String) map.get("create_by"));

        // 处理日期时间字段 - 支持带时区的格式
        if (map.get("create_time") != null) {
            document.setCreateTime(parseDateTime(map.get("create_time")));
        }

        document.setUpdateBy((String) map.get("update_by"));

        if (map.get("update_time") != null) {
            document.setUpdateTime(parseDateTime(map.get("update_time")));
        }

        if (map.get("status") != null) {
            document.setStatus(Short.valueOf(map.get("status").toString()));
        }
        document.setRemark((String) map.get("remark"));
        if (map.get("notice_id") != null) {
            document.setNoticeId(Integer.valueOf(map.get("notice_id").toString()));
        }

        return document;
    }

    /**
     * 解析日期时间字符串，支持多种格式
     */
    private LocalDateTime parseDateTime(Object dateTimeObj) {
        if (dateTimeObj == null) {
            return null;
        }

        String dateTimeStr = dateTimeObj.toString();

        try {
            // 尝试解析带时区的日期时间格式 (ISO 8601)
            OffsetDateTime offsetDateTime = OffsetDateTime.parse(dateTimeStr, ISO_OFFSET_DATE_TIME_FORMATTER);
            return offsetDateTime.toLocalDateTime();
        } catch (DateTimeParseException e1) {
            try {
                // 尝试解析不带时区的日期时间格式
                return LocalDateTime.parse(dateTimeStr, LOCAL_DATE_TIME_FORMATTER);
            } catch (DateTimeParseException e2) {
                log.warn("无法解析日期时间字符串: {}", dateTimeStr);
                // 如果两种格式都无法解析，返回null或抛出异常，根据您的需求决定
                return null;
            }
        }
    }

    /**
     * 使用新的Elasticsearch Java Client同步数据到Elasticsearch
     */
    private void syncToElasticsearch(List<NewsDocument> newsDocuments) {
        try {
            // 确保索引存在
            ensureIndexExists();

            // 准备批量索引操作
            List<BulkOperation> operations = newsDocuments.stream()
                    .map(news -> BulkOperation.of(op -> op
                            .index(idx -> idx
                                    .index("tb_news")
                                    .id(news.getNewsId().toString())
                                    .document(news)
                            )
                    ))
                    .collect(Collectors.toList());

            // 创建批量请求
            BulkRequest bulkRequest = BulkRequest.of(r -> r.operations(operations));

            // 执行批量索引
            BulkResponse bulkResponse = elasticsearchClient.bulk(bulkRequest);

            if (bulkResponse.errors()) {
                log.error("批量索引过程中发生错误");
                // 可以进一步处理错误项
                bulkResponse.items().forEach(item -> {
                    if (item.error() != null) {
                        log.error("文档 {} 索引失败: {}", item.id(), item.error().reason());
                    }
                });
            } else {
                log.debug("成功批量索引 {} 个文档", newsDocuments.size());
            }
        } catch (Exception e) {
            log.error("同步数据到Elasticsearch失败: {}", e.getMessage(), e);
            throw new RuntimeException("ES同步操作失败", e);
        }
    }

    /**
     * 确保索引存在
     */
    private void ensureIndexExists() {
        try {
            // 检查索引是否存在
            boolean exists = elasticsearchClient.indices()
                    .exists(e -> e.index("tb_news"))
                    .value();

            if (!exists) {
                // 创建索引
                elasticsearchClient.indices().create(c -> c.index("tb_news"));
                log.info("创建Elasticsearch索引: tb_news");
            }
        } catch (Exception e) {
            log.error("检查或创建索引失败: {}", e.getMessage(), e);
            throw new RuntimeException("索引操作失败", e);
        }
    }
}